package com.example.study.commen.flinkcdc;

import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.runtime.state.memory.MemoryStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import com.ververica.cdc.debezium.StringDebeziumDeserializationSchema;
import com.ververica.cdc.connectors.mysql.MySqlSource;

import java.util.Properties;

public class MySqlBinlogSourceExample {
    public static void main(String[] args) throws Exception {
        //获取flink执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);  //使用并行度1的接收器保持消息顺序

        //2.Flink-CDC将读取binlog的位置信息以状态的方式保存在CK,如果想要做到断点续传,
        // 需要从Checkpoint或者Savepoint启动程序

        //2.1 开启Checkpoint,每隔5秒钟做一次CK
        // 启用 checkpoint,设置触发间隔（两次执行开始时间间隔）
        env.enableCheckpointing(1000L * 60 * 60 * 12);//12小时
//        env.enableCheckpointing(1000L * 20);

        // 存储位置，FileSystemCheckpointStorage(文件存储)
//        env.getCheckpointConfig().setCheckpointStorage(new FileSystemCheckpointStorage("file:///c:/cdc/checkpoint/"));
        // savepoint存储位置
//        env.setDefaultSavepointDirectory("file:///c:/cdc/checkpoint/");
        // 超时时间，checkpoint没在时间内完成则丢弃
        env.getCheckpointConfig().setCheckpointTimeout(1000L*10);
        // 同时并发数量
//        env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
        // 最小间隔时间（前一次结束时间，与下一次开始时间间隔）
//        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(1000);
        // 外部checkpoint(例如文件存储)清除策略
//        env.getCheckpointConfig().setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
//        env.setStateBackend(new FsStateBackend(""));
        env.setStateBackend(new MemoryStateBackend());
//        env.setStateBackend( new FsStateBackend("file:///data/flink/checkpoints"));
       ;
        //2.2 指定CK的一致性语义
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        //2.3 设置任务关闭的时候保留最后一次CK数据
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        //2.4 指定从CK自动重启策略
        // 设置重启次数为3，重启间隔时间为1000*10ms
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 1000L * 10));

        //2.5 设置状态后端

//        env.setStateBackend(new FsStateBackend("hdfs://hadoop102:8020/flinkCDC"));

        //2.6 设置访问HDFS的用户名

//        System.setProperty("HADOOP_USER_NAME", "atguigu");

        //3.创建Flink-MySQL-CDC的Source


        Properties debeziumProperties = new Properties();
//        debeziumProperties.put("snapshot.locking.mode", "none");// do not use lock
//        debeziumProperties.setProperty("allowPublicKeyRetrieval","true");
//        debeziumProperties.put("allowPublicKeyRetrieval","true");
        //通过FlinkCDC构建SourceFunction
        SourceFunction<String> sourceFunction = MySqlSource.<String>builder()
                .hostname("8.130.21.209")
                .port(3306)
                .databaseList("study") // set captured database
                .tableList("study.token_user","study.user") // set captured table
                .username("canal")
                .password("canal")
                .deserializer(new CustomerDeserializationSchema()) // converts SourceRecord to String
//                .deserializer(new CustomerSchema())
                .debeziumProperties(debeziumProperties)
                .startupOptions(StartupOptions.initial())
                .build();


        env
                .addSource(sourceFunction)  // 获取数据
                .print();//数据打印


        //启动任务
        env.execute("flink-mysql-cdc");
    }
    /**
     * Checkpoint 其他的属性包括：
     *
     * 精确一次（exactly-once）对比至少一次（at-least-once）：你可以选择向enableCheckpointing（long interval, CheckpointingMode mode）方法中传入一个模式来选择使用两种保证等级中的哪一种。对于大多数应用来说，精确一次是较好的选择。至少一次可能与某些延迟超低（始终只有几毫秒）的应用的关联较大。
     *
     * checkpoint 超时：如果 checkpoint 执行的时间超过了该配置的阈值，还在进行中的 checkpoint 操作就会被抛弃。
     *
     * checkpoints 之间的最小时间：该属性定义在 checkpoint 之间需要多久的时间，以确保流应用在 checkpoint 之间有足够的进展。如果值设置为了 5000， 无论 checkpoint 持续时间与间隔是多久，在前一个 checkpoint 完成时的至少五秒后会才开始下一个 checkpoint。
     * 往往使用“checkpoints 之间的最小时间”来配置应用会比 checkpoint 间隔容易很多，因为“checkpoints 之间的最小时间”在 checkpoint的执行时间超过平均值时不会受到影响（例如如果目标的存储系统忽然变得很慢）。
     * 注意这个值也意味着并发 checkpoint 的数目是一。
     *
     * 并发 checkpoint 的数目: 默认情况下，在上一个 checkpoint 未完成（失败或者成功）的情况下，系统不会触发另一个checkpoint。这确保了拓扑不会在 checkpoint 上花费太多时间，从而影响正常的处理流程。 不过允许多个checkpoint 并行进行是可行的，对于有确定的处理延迟（例如某方法所调用比较耗时的外部服务），但是仍然想进行频繁的checkpoint 去最小化故障后重跑的 pipelines 来说，是有意义的。
     * 该选项不能和 “checkpoints 间的最小时间"同时使用。
     *
     * externalized checkpoints: 你可以配置周期存储 checkpoint 到外部系统中。Externalized checkpoints 将他们的元数据写到持久化存储上并且在 job 失败的时候不会被自动删除。 这种方式下，如果你的 job失败，你将会有一个现有的 checkpoint 去恢复。
     *
     * 在 checkpoint 出错时使 task 失败或者继续进行 task：他决定了在 task checkpoint 的过程中发生错误时，是否使 task 也失败，使失败是默认的行为。 或者禁用它时，这个任务将会简单的把 checkpoint错误信息报告给 checkpoint coordinator 并继续运行。
     *
     * 优先从 checkpoint 恢复（prefer checkpoint for recovery）：该属性确定 job 是否在最新的checkpoint 回退，即使有更近的 savepoint 可用，这可以潜在地减少恢复时间（checkpoint 恢复比savepoint恢复更快）。
     */
}